import beans.OrderEvent;
import beans.ReceiptEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author zkq
 * @date 2022/10/16 11:00
 */
public class TxPayMatch {
    //定义侧输出流标签
    private final static OutputTag<OrderEvent> unmatchedPays = new OutputTag<OrderEvent>("unmatched-pays"){};
    private final static OutputTag<ReceiptEvent> unmatchedReceipts = new OutputTag<ReceiptEvent>("unmatched-receipts"){};

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取订单支付事件数据
        SingleOutputStreamOperator<OrderEvent> orderStream = env.readTextFile("F:\\javasecode220620\\UserBehaviorAnalysis\\OrderTimeoutDetect\\src\\main\\resources\\OrderLog.csv")
                .map(line -> {
                    String[] split = line.split(",");
                    return new OrderEvent(new Long(split[0]), split[1], split[2], new Long(split[3]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                            @Override
                            public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                return element.getTimestamp() * 1000;
                            }
                        })
                )
                .filter(data -> !"".equals(data.getTxId())); //交易id不为空,pay事件

        //读取到账事件数据
        SingleOutputStreamOperator<ReceiptEvent> receiptStream = env.readTextFile("F:\\javasecode220620\\UserBehaviorAnalysis\\OrderTimeoutDetect\\src\\main\\resources\\ReceiptLog.csv")
                .map(line -> {
                    String[] split = line.split(",");
                    return new ReceiptEvent(split[0], split[1], new Long(split[2]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ReceiptEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<ReceiptEvent>() {
                            @Override
                            public long extractTimestamp(ReceiptEvent element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        })
                );

        //将两条流进行连接合并，进行匹配处理
        SingleOutputStreamOperator<Tuple2<OrderEvent,ReceiptEvent>> resultStream = orderStream
                .keyBy(OrderEvent::getTxId)
                .connect(receiptStream.keyBy(ReceiptEvent::getTxId))
                .process(new TxPayMatchDetect());

        resultStream.print("matched-pays");
        resultStream.getSideOutput(unmatchedPays).print("unmatched-pays");
        resultStream.getSideOutput(unmatchedReceipts).print("unmatched-receipts");

        env.execute("tx match detect job");

    }
    public static class TxPayMatchDetect extends CoProcessFunction<OrderEvent,ReceiptEvent, Tuple2<OrderEvent,ReceiptEvent>>{
        //定义状态，保存当前已经到来的订单支付事件和到账事件
        private ValueState<OrderEvent> payState;
        private ValueState<ReceiptEvent> receiptState;

        @Override
        public void open(Configuration parameters) throws Exception {
            payState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("pay",OrderEvent.class));
            receiptState = getRuntimeContext().getState(new ValueStateDescriptor<ReceiptEvent>("receipt",ReceiptEvent.class));
        }

        @Override
        public void processElement1(OrderEvent pay, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            //订单支付事件来了，判断是否已经有对应的到账事件
            ReceiptEvent receipt = receiptState.value();
            if (receipt != null ){
                //到账事件来过，输出匹配事件，清空状态
                out.collect(new Tuple2<>(pay,receipt));
                payState.clear();
                receiptState.clear();
            }else {
                //到账事件没来，注册一个定时器开始等待,5s
                ctx.timerService().registerEventTimeTimer((pay.getTimestamp()+5)*1000);
                //更新状态
                payState.update(pay);
            }
        }

        @Override
        public void processElement2(ReceiptEvent receipt, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
           //到账事件来了，判断是否有支付事件
            OrderEvent pay = payState.value();
            if (pay != null ){
                //支付事件来过，输出匹配事件，清空状态
                out.collect(new Tuple2<>(pay,receipt));
                payState.clear();
                receiptState.clear();
            }else {
                //支付事件没来，注册一个定时器开始等待,3s
                ctx.timerService().registerEventTimeTimer((receipt.getTimestamp()+3)*1000);
                //更新状态
                receiptState.update(receipt);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            //定时器触发，有可能有其中一个没来，也有可能都来了已经输出匹配了
            //判断哪个不为空，那么另一个就没来
            if (payState.value()!=null){
                ctx.output(unmatchedPays,payState.value());
            }
            if (receiptState.value()!=null){
                ctx.output(unmatchedReceipts,receiptState.value());
            }
            //清空状态
            payState.clear();
            receiptState.clear();
        }
    }
}
